package com.example.demo.controller;

import com.example.demo.componet.ReadCSV;
import com.example.demo.model.BrandJson;
import com.example.demo.model.SeansonFour;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

@RestController
@RequestMapping("/analyze")
public class CityConsumption {

    @Resource
    private JavaSparkContext sc;

    //文件地址
    @Autowired
    private String getFilePath;

    private List<SeansonFour> getDataList(){

        ReadCSV readCSV = new ReadCSV();        //"C:\\Users\\fireapproval\\Desktop\\毕业设计\\数据集\\e-commerce-sales-data-set-master\\Electronic product sales analysis.csv"

        //List<SeansonFour> dataList =
        return  readCSV.readThreeColumn2Bean(getFilePath, 1,7,11);
    }

    //List<ArrayList> returnData = new ArrayList<>();

    @CrossOrigin
    @GetMapping("/cityConsumption")
    public List<ArrayList> method1(){

        List<Integer> bejinone = new ArrayList<>(Arrays.asList(0,0))  ;  //北京季度一
        List<Integer> beijntwo = new ArrayList<>(Arrays.asList(0,0));   // 季度2
        List<Integer> beijinthree = new ArrayList<>(Arrays.asList(0,0));  //3季度
        List<Integer> beijinfour = new ArrayList<>(Arrays.asList(0,0));   //4季度

        List<Integer> shanghainone = new ArrayList<>(Arrays.asList(0,0)) ;  //伤害季度一
        List<Integer> shanghaitwo = new ArrayList<>(Arrays.asList(0,0));   // 季度2
        List<Integer> shanghaithree = new ArrayList<>(Arrays.asList(0,0));  //3季度
        List<Integer> shanghaifour = new ArrayList<>(Arrays.asList(0,0));   //4季度

        List<Integer> guangzhouone = new ArrayList<>(Arrays.asList(0,0)) ;  //广州季度一
        List<Integer> guangzhoutwo = new ArrayList<>(Arrays.asList(0,0));   // 季度2
        List<Integer> guangzhouthree = new ArrayList<>(Arrays.asList(0,0));  //3季度
        List<Integer> guangzhoufour = new ArrayList<>(Arrays.asList(0,0));   //4季度




        //System.out.println("----------cityConsumption:"+ dataList);


        List<ArrayList> returnData = new ArrayList<>();

        for (SeansonFour obj:getDataList()) {
            //城市
            String s = obj.getLocal();
            //消费
            String f = obj.getPrice();
            //月份
            String t = obj.getEventTime();

            //北京
            if ((s.equals("北京") || s == "北京")) {
                if (t.equals("01") || t.equals("02") || t.equals("03")) {
                    bejinone.add(Math.round(Float.parseFloat(f)));   //北京1季度
                }
                if (t.equals("04") || t.equals("05") || t.equals("06")) {
                    beijntwo.add(Math.round(Float.parseFloat(f)));   //北京1季度
                }
                if (t.equals("07") || t.equals("08") || t.equals("09")) {
                    beijinthree.add(Math.round(Float.parseFloat(f)));   //北京1季度
                }
                if (t.equals("01") || t.equals("02") || t.equals("03")) {
                    beijinfour.add(Math.round(Float.parseFloat(f)));   //北京1季度
                }
            }

            //广州
            if (s.equals("广东") || s == "广东") {
                if (t.equals("01") || t.equals("02") || t.equals("03")) {
                    guangzhouone.add(Math.round(Float.parseFloat(f)));   //广州1季度
                }
                if (t.equals("04") || t.equals("05") || t.equals("06")) {
                    guangzhoutwo.add(Math.round(Float.parseFloat(f)));   //
                }
                if (t.equals("07") || t.equals("08") || t.equals("09")) {
                    guangzhouthree.add(Math.round(Float.parseFloat(f)));   //
                }
                if (t.equals("10") || t.equals("11") || t.equals("12")) {
                    guangzhoufour.add(Math.round(Float.parseFloat(f)));   //
                }
            }


            //上海
            if (s.equals("上海") || s == "上海") {
                if (t.equals("01") || t.equals("02") || t.equals("03")) {
                    shanghainone.add(Math.round(Float.parseFloat(f)));   //广州1季度
                }
                if (t.equals("04") || t.equals("05") || t.equals("06")) {
                    shanghaitwo.add(Math.round(Float.parseFloat(f)));   //北京1季度
                }
                if (t.equals("07") || t.equals("08") || t.equals("09")) {
                    shanghaithree.add(Math.round(Float.parseFloat(f)));   //北京1季度
                }
                if (t.equals("10") || t.equals("11") || t.equals("12")) {
                    shanghaifour.add(Math.round(Float.parseFloat(f)));   //北京1季度
                }
            }
        }

        //System.out.println("--------------"+bejinone);
        //System.out.println("--------------"+beijntwo);
        //System.out.println("--------------"+beijinthree);
        //System.out.println("--------------"+beijinfour);

        //rdd
        JavaRDD<Integer> rdd = sc.parallelize(bejinone);
        JavaRDD<Integer> rdd1 = sc.parallelize(beijntwo);
        JavaRDD<Integer> rdd2 = sc.parallelize(beijinthree);
        JavaRDD<Integer> rdd3 = sc.parallelize(beijinfour);

        //广州
        JavaRDD<Integer> grdd = sc.parallelize(guangzhouone);
        JavaRDD<Integer> grdd1 = sc.parallelize(guangzhoutwo);
        JavaRDD<Integer> grdd2 = sc.parallelize(guangzhouthree);
        JavaRDD<Integer> grdd3 = sc.parallelize(guangzhoufour);

        //shanghai
        JavaRDD<Integer> srdd = sc.parallelize(shanghainone);
        JavaRDD<Integer> srdd1 = sc.parallelize(shanghaitwo);
        JavaRDD<Integer> srdd2 = sc.parallelize(shanghaithree);
        JavaRDD<Integer> srdd3 = sc.parallelize(shanghaifour);

        Integer breduce = rdd.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);   //北京一季度
        Integer breduce1 = rdd1.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        Integer breduce2 = rdd2.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        Integer breduce3 = rdd3.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);  //北京二季度


        Integer gbreduce = grdd.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);   //广州一季度
        Integer gbreduce1 = grdd1.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        Integer gbreduce2 = grdd2.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        Integer gbreduce3 = grdd3.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);  //广州二季度


        Integer sbreduce = srdd.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);   //sahnghai一季度
        Integer sbreduce1 = srdd1.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        Integer sbreduce2 = srdd2.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        Integer sbreduce3 = srdd3.reduce((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);  //shanghai二季度


        ArrayList one = new ArrayList();
        one.add("product");one.add("北京");one.add("广东");one.add("上海");

        ArrayList two = new ArrayList();
        two.add("第一季度");two.add(breduce);two.add(gbreduce);two.add(sbreduce);

        ArrayList three = new ArrayList();
        three.add("第二季度");three.add(breduce1);three.add(gbreduce1);three.add(sbreduce1);

        ArrayList four = new ArrayList();
        four.add("第三季度");four.add(breduce2);four.add(gbreduce2);four.add(sbreduce2);

        ArrayList five = new ArrayList();
        five.add("第四季度");five.add(breduce3);five.add(gbreduce3);five.add(sbreduce3);

        returnData.add(one);returnData.add(two);returnData.add(three);returnData.add(four);returnData.add(five);

        //消除
        rdd.unpersist();
        rdd1.unpersist();
        rdd2.unpersist();
        rdd3.unpersist();

        grdd.unpersist();
        grdd1.unpersist();
        grdd2.unpersist();
        grdd3.unpersist();

        srdd.unpersist();
        srdd1.unpersist();
        srdd2.unpersist();
        srdd3.unpersist();

        return returnData ;
    }




}
